// Copyright 2023 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "PacketQueue.h"
#include "Include/msghead.h"
#include "Log/logging.h"
#include "MemoryManagement/VmiBuffer.h"
#include <condition_variable>
#include <deque>
#include <exception>
#include <mutex>

using namespace std;

/**
 * @brief: construct
 * @param [in] multiEnq: multiEnq
 * @param [in] multiDeq: multiDeq
 * @param [in] isBlock: queue is blocked
 */
/*lint -save -e1401 */
PacketQueue::PacketQueue(bool multiEnq, bool multiDeq, bool isBlock)
    : m_totalDataSize(0),
      m_packetDeque(),
      m_lock(),
      m_conditionVariable(),
      m_multiEnq(multiEnq),
      m_multiDeq(multiDeq),
      m_block(isBlock),
      m_status(true)
{
}
/*lint -restore */
/**
 * @brief: deconstruct
 */
PacketQueue::~PacketQueue()
{
    m_status = false;
    Clear();
}

/**
 * @brief: clear queue and free memory
 */
/*lint -save -e550 */
void PacketQueue::Clear()
{
    unique_lock<mutex> lock(m_lock);
    m_conditionVariable.notify_all();
    while (!(m_packetDeque.empty())) {
        auto packet = m_packetDeque.front();
        m_packetDeque.pop_front();
        VmiFreeBuffer(packet.first); // VmiFreeBuffer会对packet.first判空
    }
    m_totalDataSize = 0;
}
/*lint -restore */
/**
 * @brief: If you want to release the communication object between threads, you should
 *         call this interface first, especially in blocking mode, and release the object
 *         after the blocked receiving thread exits.
 */
void PacketQueue::Cancel()
{
    unique_lock<mutex> lock(m_lock);
    m_status = false;
    m_conditionVariable.notify_all();
}

/**
 * @brief: resume running
 */
void PacketQueue::Continue()
{
    unique_lock<mutex> lock(m_lock);
    m_status = true;
    m_conditionVariable.notify_all();
}

/**
 * @brief: put data into data queue
 * @param [in] packet: data
 */
void PacketQueue::PutPkt(const std::pair<uint8_t *, uint32_t> &packet)
{
    if (packet.first == nullptr || packet.second == 0) {
        ERR("error: bad packet to Put, buf:%p, size:%u", packet.first, packet.second);
        VmiFreeBuffer(packet.first); // VmiFreeBuffer会检查buffer是否为nullptr
        return;
    }
    unique_lock<mutex> lock(m_lock);
    if (m_status) {
        m_totalDataSize += packet.second;
        m_packetDeque.push_back(packet);
        if (m_block) {
            m_conditionVariable.notify_one();
        }
    } else {
        VmiFreeBuffer(packet.first); // VmiFreeBuffer会检查packet.first是否为nullptr
    }
}

/**
 * @brief: get a data from queue
 * @return : a pair that consist of by pointer and size of data packet,
 *           or a pair consist of nullptr and 0 if there are no data packet in the queue
 */
std::pair<uint8_t *, uint32_t> PacketQueue::GetNextPkt()
{
    std::pair<uint8_t *, uint32_t> packet(nullptr, 0);
    unique_lock<mutex> lock(m_lock);
    if (m_status) {
        if (m_block) {
            m_conditionVariable.wait(
                lock, [this]() -> bool { return !m_status || !m_packetDeque.empty(); });
        }
        if (m_status && !m_packetDeque.empty()) {
            packet = m_packetDeque.front();
            m_packetDeque.pop_front();
            m_totalDataSize -= packet.second;
        }
    }
    return packet;
}

/**
 * @brief: get a data from queue with timeout
 * @param [in] timeoutMillis: unit is ms
 * @return : a pair that consist of by pointer and size of data packet,
 *           or a pair consist of nullptr and 0 if there are no data packet in the queue
 */
std::pair<uint8_t *, uint32_t> PacketQueue::GetNextPktWait(int timeoutMillis)
{
    std::pair<uint8_t *, uint32_t> packet(nullptr, 0);
    unique_lock<mutex> lock(m_lock);
    if (m_status) {
        if (m_block) {
            m_conditionVariable.wait_for(lock, std::chrono::milliseconds(timeoutMillis),
                [this]() -> bool { return !m_status || !m_packetDeque.empty(); });
        }
        if (m_status && !m_packetDeque.empty()) {
            packet = m_packetDeque.front();
            m_packetDeque.pop_front();
            m_totalDataSize -= packet.second;
        }
    }

    return packet;
}

/**
 * @brief: get queue size
 * @return : number of items in the queue
 */
size_t PacketQueue::GetNumItems()
{
    unique_lock<mutex> lock(m_lock);
    return m_packetDeque.size();
}

/**
 * @brief: get total size
 * @return : total data size
 */
uint32_t PacketQueue::GetTotalSize()
{
    return m_totalDataSize;
}
